package pku;

import java.io.DataInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedList;
import java.util.List;

/**
 * 消费者
 */

public class Consumer {
	List<String> topics = new LinkedList<>();
	int readPos = 0;
	String queue;
	String[] filenames = null;
	int offset = 0;
	boolean flag = true;
	// DataInputStream dis = null;

	FileInputStream fileInputStream = null;
	FileChannel channel = null;
	ByteBuffer buffer = null;
	List<String> list = new ArrayList<>();

	// 将消费者订阅的topic进行绑定
	public void attachQueue(String queueName, Collection<String> t) throws Exception {
		if (queue != null) {
			throw new Exception("只允许绑定一次");
		}
		queue = queueName;
		topics.addAll(t);

		for (String topic : topics) {
			File file = new File(Constant.PATH + topic);
			if (file.exists()) {
				String[] fns = file.list();
				List<String> ls = new ArrayList<>();
				for (String s : fns) {
					list.add(topic + "/" + s);
				}
			}
		}
	}

	// 每次消费读取一个message
	public ByteMessage poll() {

		if (readPos > list.size()) {
			return null;
		}

		try {

			short mark = buffer.getShort();
			KeyValue headers = new DefaultKeyValue();
			for (int i = 0; i < 16; i++) {
				if (MyUtilTool.getBit(mark, i)) {
					switch (i) {
					case 0:
					case 8:
					case 9:
					case 10:
						int value = buffer.getInt();
						headers.put(MessageHeaderConstant.getHeaderInverseMapValue(i), value);
						break;
					case 13:
					case 14:
						double value1 = buffer.getDouble();
						headers.put(MessageHeaderConstant.getHeaderInverseMapValue(i), value1);
						break;
					case 2:
					case 4:
					case 6:
					case 7:
						long value2 = buffer.getLong();
						headers.put(MessageHeaderConstant.getHeaderInverseMapValue(i), value2);
						break;
					case 1:
					case 3:
					case 5:
					case 11:
					case 12:
					case 15:
						int len = buffer.getInt();
						byte[] byteLen = new byte[len];
						buffer.get(byteLen);
						String value4 = new String(byteLen);
						headers.put(MessageHeaderConstant.getHeaderInverseMapValue(i), value4);
						break;
					}
				}
			}

			int bodylen = buffer.getInt();
			byte[] body = new byte[bodylen];
			buffer.get(body);

			DefaultMessage message = new DefaultMessage(body);
			message.setHeaders(headers);
			return message;

		} catch (Exception e) {
			try {
				if (readPos >= list.size()) {
					return null;
				}
				File file1 = new File(Constant.PATH + list.get(readPos++));
				// dis = new DataInputStream(new FileInputStream(file1));
				if (buffer!=null) {
					buffer.clear();
				}
				fileInputStream = new FileInputStream(file1);
				channel = fileInputStream.getChannel();
				buffer = ByteBuffer.allocate((int) channel.size());
				channel.read(buffer);
				buffer.flip();
				short mark = buffer.getShort();
				KeyValue headers = new DefaultKeyValue();
				for (int i = 0; i < 16; i++) {
					if (MyUtilTool.getBit(mark, i)) {
						switch (i) {
						case 0:
						case 8:
						case 9:
						case 10:
							int value = buffer.getInt();
							headers.put(MessageHeaderConstant.getHeaderInverseMapValue(i), value);
							break;
						case 13:
						case 14:
							double value1 = buffer.getDouble();
							headers.put(MessageHeaderConstant.getHeaderInverseMapValue(i), value1);
							break;
						case 2:
						case 4:
						case 6:
						case 7:
							long value2 = buffer.getLong();
							headers.put(MessageHeaderConstant.getHeaderInverseMapValue(i), value2);
							break;
						case 1:
						case 3:
						case 5:
						case 11:
						case 12:
						case 15:
							int len = buffer.getInt();
							byte[] byteLen = new byte[len];
							buffer.get(byteLen);
							String value4 = new String(byteLen);
							headers.put(MessageHeaderConstant.getHeaderInverseMapValue(i), value4);
							break;
						}
					}
				}

				int bodylen = buffer.getInt();
				byte[] body = new byte[bodylen];
				buffer.get(body);

				DefaultMessage message = new DefaultMessage(body);
				message.setHeaders(headers);
				return message;
			} catch (Exception e1) {
				e1.printStackTrace();
				return null;
			}
		}

	}

}
